#include "rpcprovider.h"
#include "mprpcapplication.h"
#include "logger.h"

#include <google/protobuf/descriptor.h>
#include <string>
#include <functional> // for std::bind

#include "rpcheader.pb.h" // for mprpc:: message RpcHeader
#include <zookeeperutil.h>

/*
    service_name => 对应 serviceDescriptor
                                            => 对应service * 服务对象
                                            => method_name => method方法
 */

//  NotifyService()是框架提供给外部使用的， 可以用来发布rpc方法的函数接口
void RpcProvider::NotifyService(google::protobuf::Service *service)
{
    if (!service)
    {
        return;
    }
    ServiceInfo service_info;
    // 1.获取服务对象的描述信息
    const google::protobuf::ServiceDescriptor *pservicedesc = service->GetDescriptor();
    // 获取服务名称
    const std::string service_name = pservicedesc->name();
    // 获取方法数量
    int method_count = pservicedesc->method_count();

    // std::cout << "service_name:" << service_name << std::endl;
    LOG_INFO("service_name:%s", service_name.c_str());
    // 遍历所有方法
    for (int i = 0; i < method_count; i++)
    {
        // 2.获取服务对象指定【下标】方法的描述信息(抽象描述) UserService服务名 Login方法名
        const google::protobuf::MethodDescriptor *pmethodDesc = pservicedesc->method(i);
        std::string method_name = pmethodDesc->name(); // 得到 该方法的名称
        service_info.m_methodMap.insert({method_name, pmethodDesc});

        LOG_INFO("method_name:%s", method_name.c_str());
    }

    service_info.m_service = service;
    m_serviceMap.insert({service_name, service_info});
}

// 启动rpc服务节点， 开始提供rpc远程网络调用 服务
void RpcProvider::run()
{
    std::string ip = MprpcApplication::GetInstance().GetConig().load("rpcserverip");
    uint16_t port = static_cast<uint16_t>(std::stoul(MprpcApplication::GetInstance().GetConig().load("rpcserverport")));
    bool isIpv6 = false;
    muduo::net::InetAddress address(ip, port, isIpv6); // 监听接口和端口

    // 创建TcpServer对象
    muduo::net::TcpServer server(&m_eventLoop, address, "RpcProvider");
    // 绑定连接回调和消息读写回调方法
    server.setConnectionCallback(std::bind(&RpcProvider::onConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcProvider::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

    // 设置muduo库的线程数量
    server.setThreadNum(2); // 设置muduo库的线程数量

    // 把当前rpc节点上要发布的服务全部注册到zookeeper上， 让rpc client可以从zookeeper上发现服务
    // 默认session的timeout 为 30s 每三分之一的timeout时间发送ping消息到zookeeper server端
    ZkClient zkCli;
    zkCli.Start();
    //service_name为永久性节点， method_name 为ZOO_EPHEMERAL 表示创建临时节点
    for (auto &sp : m_serviceMap)
    {
        // path为：/service_name
        std::string service_path = '/' + sp.first;
        zkCli.Create(service_path.c_str(), nullptr, 0);
        for (auto &mp : sp.second.m_methodMap)
        {
            // path为: /service_name/method_name 存储当前这个rpc服务节点的url: ip:port
            std::string method_path = service_path + '/' + mp.first;
            char method_path_data[128] = {0};
            sprintf(method_path_data, "%s:%d", ip.c_str(), port);
            // 方法节点为临时节点ephemeral
            zkCli.Create(method_path.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);
        }        
    }
    std::cout << "RpcProvider start service at ip:" << ip << " port:" << port << std::endl;
    server.start();     // 启动服务器
    m_eventLoop.loop(); // 进入事件循环以等待事件发生
}

// 连接回调
void RpcProvider::onConnection(const muduo::net::TcpConnectionPtr &conn)
{
    // 连接回调只负责把断开设置为conn->shutdown()连接关闭
    if (conn->connected())
    {
        std::cout << "New connection from " << conn->peerAddress().toIpPort() << std::endl;
    }
    else
    {
        // 和 rpc client 的连接断开
        std::cout << "Connection lost from " << conn->peerAddress().toIpPort() << std::endl;
        conn->shutdown();
    }
}

/*
    在框架内部， RpcProvider 和  RpcConsumer协商好， 两者之间通信用的protobuf数据类型
    header_size(4个字节） + header_str + args_str
*/
// 消息读写回调
void RpcProvider::onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp time)
{
    // 1.接收消息
    std::string recv_buf(buf->retrieveAllAsString());
    // 2.解析头部信息
    uint32_t header_size = 0;
    memcpy(&header_size, recv_buf.data(), 4);

    // 3.反序列化头部
    std::string rpc_header_str = recv_buf.substr(4, header_size);
    mprpc::RpcHeader rpcHeader;
    if (!rpcHeader.ParseFromString(rpc_header_str))
    {
        // 数据头反序列化失败
        std::cout << "rpc_header_str:" << rpc_header_str << "parse error!" << std::endl;
        return;
    }
    // 4.提取服务名、方法名和参数大小
    std::string serviceName = rpcHeader.service_name();
    std::string methodName = rpcHeader.method_name();
    uint32_t argsSize = rpcHeader.args_size();

    // 5.提取参数字符串
    std::string args_str = recv_buf.substr(4 + header_size, argsSize);

    // 6.调试信息输出：
    std::cout << "================================================";
    std::cout << "header_size: " << header_size << std::endl;
    std::cout << "rpc_header_str: " << rpc_header_str << std::endl;
    std::cout << "serviceName: " << serviceName << std::endl;
    std::cout << "methodName: " << methodName << std::endl;
    std::cout << "argsSize: " << argsSize << std::endl;
    std::cout << "args_str: " << args_str << std::endl;
    std::cout << "================================================";

    // 7.查找服务和方法
    auto it = m_serviceMap.find(serviceName);
    if (it == m_serviceMap.end())
    {
        std::cout << serviceName << " is not exist!" << std::endl;
        return;
    }
    auto mit = it->second.m_methodMap.find(methodName);
    if (mit == it->second.m_methodMap.end())
    {
        std::cout << serviceName << ":" << methodName << " is not exist!" << std::endl;
        return;
    }
    google::protobuf::Service *service = it->second.m_service;
    const google::protobuf::MethodDescriptor *method = mit->second;

    // 8.创建请求和响应实例
    std::unique_ptr<google::protobuf::Message> request(service->GetRequestPrototype(method).New());
    // 8.1并尝试从参数字符串反序列化请求内容
    // 如果 ParseFromString 成功执行，那么 request 对象将会包含客户端发送的所有参数，可以用来调用相应的服务方法
    if (!request->ParseFromString(args_str))
    {
        std::cout << "request parse error, content:" << args_str << std::endl;
        return;
    }
    std::unique_ptr<google::protobuf::Message> response(service->GetResponsePrototype(method).New());

    // 给rpc服务UserServiceRpc::CallMethod（）绑定一个Closure的回调函数
    google::protobuf::Closure *done =
        google::protobuf::NewCallback<RpcProvider, const muduo::net::TcpConnectionPtr &, google::protobuf::Message *>(this, &RpcProvider::SendRpcResponse, conn, response.get());
    // 9.执行远程调用 (在框架上根据远端rpc请求， 调用当前rpc节点上发布的方法)
    service->CallMethod(method, nullptr, request.get(), response.get(), done);
}

// google::protobuf::closure* 回调操作, 用于序列化rpc的响应和网络发送
void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response)
{
    std::string response_str;
    if (!response->SerializeToString(&response_str)) // response进行序列化
    {
        std::cout << "Serialize response_str to string error!" << std::endl;
    }
    else
    {
        // 序列化成功后， 通过网络把rpc方法执行的结果发送给远程过程调用的调用方
        conn->send(response_str);
    }

    conn->shutdown(); // 模拟http的短链接服务， 由rpcprovider主动断开连接
}
